package com.example.listener;

import com.alibaba.fastjson.JSON;
import com.example.latch.DbLeaderLeaderLatch;
import com.example.latch.LeaderLatchListener;
import com.example.message.ChangeData;
import com.example.message.Message;
import com.example.po.SysListener;
import com.example.service.SysListenerService;
import com.example.utils.JSONUtils;
import io.debezium.config.Configuration;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import lombok.Data;
import org.slf4j.Logger;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Data
@Service
public class SysDebeziumListener implements InitializingBean, DisposableBean , ApplicationContextAware , LeaderLatchListener {

    Logger logger = org.slf4j.LoggerFactory.getLogger(getClass());

    @Value("${debezium.datasource.server.name}")
    private String servername;

    @Value("${debezium.datasource.username}")
    private String username;

    @Value("${debezium.datasource.password}")
    private String password;

    @Value("${debezium.datasource.db}")
    private String db;

    @Value("${debezium.datasource.host}")
    private String host;

    @Value("${debezium.datasource.port}")
    private int port;

    @Value("${debezium.datasource.serverTimezone}")
    private String serverTimezone;

    @Value("${server.port}")
    private int serverId;

    private ExecutorService executor;

    private DebeziumEngine engine;

    private ApplicationContext applicationContext;

    private List<SysListener> sysListeners = new ArrayList<>();

    @Resource
    private SysListenerService sysListenerService;

    @Resource
    private JmsTemplate jmsTemplate;

    private DbLeaderLeaderLatch dbLeaderLeaderLatch;

    @Override
    public void destroy() throws Exception {
       this.stopServer();
        this.dbLeaderLeaderLatch.stop();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.executor = Executors.newCachedThreadPool();
        this.dbLeaderLeaderLatch = new DbLeaderLeaderLatch(this.applicationContext, this);
        this.dbLeaderLeaderLatch.start();
    }

    //启动方法
    private void startServer() {
        Configuration config = Configuration.create().with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
                .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
                .with("offset.storage.file.filename", "/tmp/offset.dat")
                .with("offset.flush.interval.ms", 60000)
                .with("name", "my-sql-connector")
                .with("database.hostname", host)
                .with("database.port", port)
                .with("database.user", username)
                .with("database.password", password)
                .with("database.connectionTimeZone", serverTimezone)
                .with("database.server.id", serverId)
                .with("database.include.list", db)
                .with("snapshot.mode", "initial")
                .with("database.server.name", servername)
                .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
                .with("database.history.file.filename", "/tmp/history.dat")
                .build();
        this.engine = DebeziumEngine.create(Json.class).using(config.asProperties()).notifying(record -> {
            receiveChangeEvent(record.value());
        }).build();
        this.executor.execute(this.engine);
        logger.warn("Debezium server start success ...");
    }

    //收到消息先预处理
    public void receiveChangeEvent(String value) {
        if (Objects.nonNull(value)) {
            Map<String, Object> payload = JSONUtils.getPayload(value);
            String handleType = JSONUtils.getHandleType(payload);
            if (!("NONE".equals(handleType) || "READ".equals(handleType))) {
                ChangeData changeData = JSONUtils.getChangeData(payload);
                Map<String, Object> data;
                if ("DELETE".equals(handleType)) {
                    data = changeData.getBefore();
                } else {
                    data = changeData.getAfter();
                }
                Message message = new Message();
                message.setData(data);
                message.setDbType("MYSQL");
                message.setDatabase(String.valueOf(changeData.getSource().get("db")));
                message.setTable(String.valueOf(changeData.getSource().get("table")));
                message.setHandleType(handleType);
                this.executor.execute(new SysDataChangeHander(message));
            }
        }
    }

    //停止方法
    private void stopServer() throws Exception {
        this.engine.close();
        this.executor.shutdownNow();
    }

    @Override
    public void isLeader() {
        this.startServer();
    }

    //定时执行拉取监听器列表
    @Scheduled(cron ="0/30 * * * * ?")
    public void refreshListeners() {
        this.sysListeners = sysListenerService.list();
    }

    //自定义的处理器
    class SysDataChangeHander implements Runnable {

        private Message message;

        public SysDataChangeHander(Message message) {
            this.message = message;
        }

        @Override
        public void run() {
            if (!getSysListeners().isEmpty()) {
                for (SysListener sysListener : getSysListeners()) {
                    if (sysListener.getDbScheme().equalsIgnoreCase(message.getDatabase()) && sysListener.getDbTable().equalsIgnoreCase(message.getTable()) && sysListener.getDbEvent().equalsIgnoreCase(message.getHandleType())) {
                        Map<String, Object> data = message.getData();
                        jmsTemplate.convertAndSend(sysListener.getQueueName(), JSON.toJSONString(data));
                    }
                }
            }
        }
    }
}
